<?php

namespace App\Models\Service;

use App\Models\Common\ApiResponse;
use Illuminate\Support\Arr;
use Workerman\Connection\TcpConnection;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
use Workerman\Mqtt\Client;
use App\Jobs\MqttMessageHandler;
use App\Models\Common\Constant;

class Mqtt
{
    public static function getConfig(): array
    {
        $mqttConfig = config('services.mqtt') ?? [];

        $address = strval(\Arr::get($mqttConfig, 'address'));
        $username = strval(\Arr::get($mqttConfig, 'username'));
        $password = strval(\Arr::get($mqttConfig, 'password'));
        $keepalive = intval(\Arr::get($mqttConfig, 'keepalive')) ?: 30;
        $connectTimeout = intval(\Arr::get($mqttConfig, 'connect_timeout')) ?: 10;
        $reconnectPeriod = intval(\Arr::get($mqttConfig, 'reconnect_period')) ?: 3;
        $debug = boolval(\Arr::get($mqttConfig, 'debug'));

        if (!$address) {
            throw new \Exception('mqtt服务地址未配置');
        }

        if (stripos($address, 'mqtt://') === false) {
            $address = 'mqtt://' . $address;
        }

        return compact('address', 'username', 'password', 'keepalive', 'connectTimeout', 'reconnectPeriod', 'debug');
    }

    public static function logging(string $logContent, $logDat = [])
    {
        if (!is_array($logDat)) {
            $logDat = \Arr::wrap($logDat);
        }
        \Log::channel('mqtt')->info($logContent, $logDat);
    }

    public static function on()
    {
        $worker = new Worker('http://localhost:8099/mqtt/service/');
        $logFileName = 'logs/service/worker-master-' . now()->toDateString() . '.log';
        Worker::$logFile = storage_path($logFileName);

        $worker->onWorkerStart = function () use ($worker) {

            $mqttConfig = Mqtt::getConfig();
            $address = \Arr::get($mqttConfig, 'address');
            unset($mqttConfig['address']);
            $mqtt = new Client($address, $mqttConfig);

            $mqtt->onConnect = function (Client $mqtt) {
                foreach (VisionZenith::SUBSCRIBE_TOPICS as $rowTopic) {
                    $mqtt->subscribe($rowTopic);
                }
            };

            $worker->onMessage = function (TcpConnection $tcpConnection, Request $request) use ($mqtt) {

                try {

                    self::logging('http服务请求', ['request' => $request->post(), 'ip' => $tcpConnection->getRemoteIp()]);

                    if (!in_array($tcpConnection->getRemoteIp(), ['127.0.0.1', '[::1]', 'localhost'])) {
                        throw new \Exception('仅限本地数据交互:' . $tcpConnection->getRemoteIp());
                    }

                    $postBody = $request->post();
                    $messages = Arr::get($postBody, 'messages');
                    $topic = Arr::get($postBody, 'topic');
                    $topic = trim($topic, '/');
                    $topic = '/' . $topic;

                    if ($topic && $messages && is_array($messages)) {
                        foreach ($messages as $message) {
                            if ($jsonMessage = @json_encode($message)) {
                                $mqtt->publish($topic, $jsonMessage);
                                self::logging($topic . ':发布成功', $message);
                            }
                        }
                    }

                    $tcpConnection->send(@json_encode(ApiResponse::return('', '', ApiResponse::RESP_CODE_SUCCESS)));
                } catch (\Throwable $exception) {
                    self::logging($exception->getMessage(), @json_decode($request->rawBody(), true) ?: []);
                    $tcpConnection->send(@json_encode(ApiResponse::return($exception->getMessage(), '', ApiResponse::RESP_CODE_MQTT_SERVICE_ERROR)));
                }
            };

            $mqtt->onMessage = function (string $topic, string $content, Client $mqtt) {

                $fmtContent = @json_decode($content, true) ?: [];
                $logTag = $topic . ':';
                if (is_array($fmtContent) && $fmtContent) {
                    if (in_array($topic, VisionZenith::SUBSCRIBE_TOPICS, true)) {
                        MqttMessageHandler::dispatch(VisionZenith::class, $fmtContent)->onQueue(Constant::QUEUE_DEFAULT);
                        $logTag .= '回执处理';
                    } else {
                        $logTag .= '回执丢弃';
                    }
                    self::logging($logTag, $fmtContent);
                } else {
                    self::logging($logTag . '回执废弃', ['raw_content' => $content]);
                }
            };

            $mqtt->onError = function (\Exception $exception) {

            };

            $mqtt->onClose = function () {

            };

            $mqtt->connect();
        };

        $worker->onWorkerExit = function ($worker, $statusCode) {
            Worker::stopAll();
        };
        Worker::runAll();
    }
}
